package com.tt.job.zookeeper;

import com.tt.job.JobInfo;
import com.tt.job.api.Job;
import com.tt.job.api.schedule.JobSchedule;
import com.tt.job.common.util.JsonUtil;
import com.tt.job.register.JobChangeListener;
import com.tt.job.register.JobRegistryCenter;
import com.tt.job.register.LeaderChangeListener;
import com.tt.job.register.PathConst;
import com.tt.job.register.ScheduleChangeListener;
import com.tt.job.util.JobUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * zookeeper注册中心服务
 */
@Service
public class ZookeeperRegistryCenter implements JobRegistryCenter {
    private final static Logger logger = LoggerFactory.getLogger(ZookeeperRegistryCenter.class);

    //@Autowired
    //private ZookeeperBeanFactory beanFactory;
    @Autowired
    private CuratorFramework client;
    private PathChildrenCache jobCache;
    private PathChildrenCache jobScheduleCache;
    @Autowired
    private LeaderLatch leaderLatch;
    @Autowired
    private ZookeeperSettings settings;

    @Override
    public void addJob(String jobClass, Job job) {
        try {
            //注册job节点
            String fullPath = getJobFullPath(jobClass, job.getType().toString());
            persist(fullPath, JsonUtil.encodeString(job));
        } catch (Exception e) {
            logger.error("add job to zookeeper error.");
        }
    }

    @Override
    public void addJobListener(JobChangeListener listener) {
        //
        String jobPath = getJobPath();
        try {
            synchronized (client) {
                if (jobCache == null) {
                    jobCache = new PathChildrenCache(client, jobPath, true);
                    jobCache.start();
                }
            }
            jobCache.getListenable().addListener(new ZookeeperJobChangeExecutor(listener));
        } catch (Exception e) {
            logger.error("add job listener to zookeeper error.", e);
        }

    }

    @Override
    public void addJobSchedule(JobSchedule schedule) {
        //获取所有的schedule节点
        String jobSchedulePath = getJobSchedulePath();
        String path = String.format("%s/%s-%s-%s", jobSchedulePath, schedule.getType(), schedule.getGroup(), schedule.getName());
        try {
            if (!isExisted(path)) {
                persist(path, JsonUtil.encodeString(schedule));
            }
        } catch (Exception e) {
            logger.error("add schedule to zookeeper error.", e);
        }
    }

    @Override
    public void addJobScheduleListener(ScheduleChangeListener listener) {
        //
        String schedulePath = getJobSchedulePath();
        try {
            synchronized (client) {
                if (jobScheduleCache == null) {
                    jobScheduleCache = new PathChildrenCache(client, schedulePath, true);
                    jobScheduleCache.start();
                }
            }
            jobScheduleCache.getListenable().addListener(new ZookeeperScheduleChangeExecutor(listener));
        } catch (Exception e) {
            logger.error("add job schedule listener to zookeeper error.", e);
        }
    }

    @Override
    public void deleteSchedule(String type, String group, String name) {
        String jobSchedulePath = getJobSchedulePath();
        String path = String.format("%s/%s-%s-%s", jobSchedulePath, type, group, name);
        try {
            remove(path);
        } catch (Exception e) {
            logger.error("delete schedule from zookeeper error.", e);
        }
    }

    @Override
    public List<JobSchedule> getJobSchedules() {
        //获取所有的schedule节点
        String jobSchedulePath = getJobSchedulePath();
        try {
            List<String> schedules = getChildren(jobSchedulePath);
            List<JobSchedule> results = new ArrayList<>(schedules.size());
            for (String schedule : schedules) {
                String[] scheduleInfos = schedule.split("-");
                if (scheduleInfos.length != 3) {
                    logger.error("schedule path {} are broken!", schedule);
                    continue;
                }
                byte[] data = client.getData().forPath(jobSchedulePath + "/" + schedule);
                JobSchedule jobSchedule = JobUtils.getScheduleInstance(scheduleInfos[0], data);
                results.add(jobSchedule);
            }
            return results;
        } catch (Exception e) {
            logger.error("get schedules from zookeeper error. path:{}", jobSchedulePath, e);
        }
        return null;
    }

    @Override
    public Map<String, JobInfo> getJobs() {
        //获取所有的job节点
        String jobPath = getJobPath();
        Map<String, JobInfo> jobMap = new HashMap<>();
        try {
            if (!isExisted(jobPath)) {
                persist(jobPath, "");
            }
            List<String> jobs = getChildren(jobPath);
            for (String jobInfo : jobs) {
                String[] jobInfos = jobInfo.split("-");
                if (jobInfos.length != 2) {
                    logger.error("job path {} are broken!", jobInfo);
                    continue;
                }
                String jobClass = jobInfos[1];
                byte[] data = client.getData().forPath(jobPath + "/" + jobInfo);
                jobMap.put(jobInfos[1], JobUtils.getJobInstance(jobInfos[0], jobClass, data));
            }
            return jobMap;
        } catch (Exception e) {
            logger.error("get jobs from zookeeper error. path:{}", jobPath, e);
        }
        return jobMap;
    }

    @Override
    public void registryJobClient(String jobClass, String jobType, String localPid) {
        String jobPath = getJobFullPath(jobClass, jobType);
        String fullPath = String.format("%s%s/%s", jobPath, PathConst.JOB_CLIENTS, localPid);
        try {
            ephemeral(fullPath, "");
        } catch (Exception e) {
            logger.error("register job client to zookeeper error.");
        }
    }

    private String getJobPath() {
        return settings.getRoot() + PathConst.JOB;
    }

    public List<String> getChildren(String rootPath) throws Exception {
        return this.client.getChildren().forPath(rootPath);
    }

    private String getJobSchedulePath() {
        return settings.getRoot() + PathConst.SCHEDULE;
    }

    private String getJobFullPath(String jobClass, String jobType) {
        String jobPath = getJobPath();
        return String.format("%s/%s-%s", jobPath, jobType, jobClass);
    }

    @Override
    public void addLeaderListener(LeaderChangeListener listener) {
        try {
            if (leaderLatch != null) {
                leaderLatch.addListener(new LeaderLatchListener() {
                    @Override
                    public void isLeader() {
                        listener.notifyLeader();
                    }

                    @Override
                    public void notLeader() {
                        listener.notifyNotLeader();
                    }
                });
            }
        } catch (Exception e) {
            logger.error("add listener to zookeeper register center error.", e);
        }
    }

    @Override
    public void close() {
        client.close();
        try {
            leaderLatch.close();
        } catch (IOException e) {
            logger.error("close leader latch error.", e);
        }
    }

    @Override
    public boolean ephemeral(String key, String value) throws Exception {
        String path = this.client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key, value.getBytes());
        return path != null;
    }

    @Override
    public String get(String key) throws Exception {
        return new String(client.getData().forPath(key));
    }

    @Override
    public Object getRawClient() {
        return client;
    }

    @Override
    public boolean isExisted(String key) throws Exception {
        return this.client.checkExists().forPath(key) != null;
    }

    @Override
    public boolean leading() {
        try {
            leaderLatch.start();
            return leaderLatch.await(10, TimeUnit.SECONDS);
        } catch (Exception e) {
            logger.error("zookeeper leading failed!", e);
        }
        return false;
    }

    @Override
    public void persist(String key, String value) throws Exception {
        this.client.create().creatingParentsIfNeeded().forPath(key, value.getBytes());
    }

    @Override
    public void remove(String key) throws Exception {
        if (this.client.checkExists().forPath(key) != null) {
            this.client.delete().forPath(key);
        }
    }

    @Override
    public void start() throws Exception {
        try {
            client.start();
            //检查并创建必要的节点
            if (this.client.checkExists().forPath(settings.getRoot()) == null) {
                this.client.create().forPath(settings.getRoot());
            }
        } catch (Exception e) {
            logger.error("connect the zookeeper register center error.", e);
        }
    }

    @Override
    public void update(String key, String value) throws Exception {
        this.client.setData().forPath(key, value.getBytes());
    }

    public LeaderLatch getLeaderLatch() {
        return leaderLatch;
    }
}
